package com.atguigu.chapter07.state;

import com.atguigu.bean.WaterSensor;
import com.atguigu.util.MyUtil;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/15 14:02
 */
public class Flink04_State_Key_List {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment
            .getExecutionEnvironment()
            .setParallelism(3);
        env
            .socketTextStream("hadoop162", 9999)
            .map(value -> {
                String[] datas = value.split(",");
                return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
                
            })
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                
                private ListState<Integer> top3VcState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    top3VcState = getRuntimeContext()
                        .getListState(new ListStateDescriptor<Integer>("top3VcState", Integer.class));
                }
                
                @Override
                public void processElement(WaterSensor value,
                                           Context ctx,
                                           Collector<String> out) throws Exception {
                    top3VcState.add(value.getVc());
                    
                    // 排序取前3
                    Iterable<Integer> vcIt = top3VcState.get();
                    List<Integer> vcs = MyUtil.toList(vcIt);
                    /*vcs.sort(new Comparator<Integer>() {
                        @Override
                        public int compare(Integer o1, Integer o2) {
                            return o2.compareTo(o1);
                        }
                    });*/
                    //                    vcs.sort((o1, o2) -> o2.compareTo(o1));
                    vcs.sort(Comparator.reverseOrder());  // java中的排序, 都是在原集合(数组)只就地排序, 排除成功之后, 原来的集合就会有序了.
                    
                    if (vcs.size() > 3) {
                        vcs.remove(vcs.size() - 1);
                    }
                    
                    out.collect(ctx.getCurrentKey() + ":" + vcs.toString());
                    
                    top3VcState.update(vcs);
                    
                }
            })
            .print();
        
        env.execute();
        
    }
}
